package org.study.r2dbc.boot.service;

import io.r2dbc.spi.Row;
import io.r2dbc.spi.RowMetadata;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.stereotype.Service;
import org.springframework.transaction.reactive.TransactionalOperator;
import org.study.r2dbc.boot.model.User;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.math.BigDecimal;
import java.util.List;
import java.util.function.BiFunction;

/**
 * 作用描述
 *
 * @author doveylovey
 * @version v1.0.0
 * @email 1135782208@qq.com
 * @date 2025年04月02日
 */
@Slf4j
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
@Service
public class UserServiceImplByDatabaseClient implements UserServiceByDatabaseClient {
    private final DatabaseClient databaseClient;
    private final TransactionalOperator transactionalOperator;

    @Override
    public Mono<User> findUserById(Long id) {
        return databaseClient.sql("SELECT * FROM t_user WHERE id = :id")
                .bind("id", id)
                .map((row, rowMetadata) -> {
                    User user = new User();
                    user.setId(row.get("id", Long.class));
                    user.setName(row.get("name", String.class));
                    user.setAge(row.get("age", Integer.class));
                    user.setEmail(row.get("email", String.class));
                    user.setBalance(new BigDecimal(row.get("balance", String.class)));
                    user.setActive(row.get("active", Boolean.class));
                    return user;
                })
                .one();
    }

    @Override
    public Flux<User> findAllUsers1() {
        return databaseClient.sql("SELECT * FROM t_user")
                .map((row, rowMetadata) -> {
                    User user = new User();
                    user.setId(row.get("id", Long.class));
                    user.setName(row.get("name", String.class));
                    user.setAge(row.get("age", Integer.class));
                    user.setEmail(row.get("email", String.class));
                    user.setBalance(new BigDecimal(row.get("balance", String.class)));
                    user.setActive(row.get("active", Boolean.class));
                    return user;
                })
                .all();
    }

    @Override
    public Flux<User> findAllUsers2() {
        return databaseClient.sql("SELECT * FROM t_user")
                // .map(userRowMapper)
                .map(new UserRowMapper())
                .all();
    }

    @Override
    public Mono<Long> insertUser1(User user) {
        return databaseClient.sql("INSERT INTO t_user(name, age, email, balance, active) VALUES(:name, :age, :email, :balance, :active)")
                .bind("name", user.getName())
                .bind("age", user.getAge())
                .bind("email", user.getEmail())
                .bind("balance", user.getBalance())
                .bind("active", user.getActive())
                .filter(statement -> statement.returnGeneratedValues("id"))
                .fetch()
                .first()
                .map(r -> (Long) r.get("id"));
    }

    @Override
    public Mono<Long> insertUser2(User user) {
        return databaseClient.sql("INSERT INTO t_user(id, name, age, email, balance, active) VALUES(:id, :name, :age, :email, :balance, :active)")
                .bind("id", user.getId())
                .bind("name", user.getName())
                .bind("age", user.getAge())
                .bind("email", user.getEmail())
                .bind("balance", user.getBalance())
                .bind("active", user.getActive())
                .fetch()
                .first()
                .map(r -> (Long) r.get("id"));
    }

    @Override
    public Mono<Long> updateUser(User user) {
        return databaseClient.sql("UPDATE t_user SET name = :name, age = :age, email = :email, balance = :balance, active = :active WHERE id = :id")
                .bind("name", user.getName())
                .bind("age", user.getAge())
                .bind("email", user.getEmail())
                .bind("balance", user.getBalance())
                .bind("active", user.getActive())
                .bind("id", user.getId())
                .fetch()
                .rowsUpdated();
    }

    @Override
    public Mono<Long> deleteUser(Long id) {
        return databaseClient.sql("DELETE FROM t_user WHERE id = :id")
                .bind("id", id)
                .fetch()
                .rowsUpdated();
    }

    @Override
    public Flux<User> findUsersByNameAndEmail1(String name, String email) {
        return databaseClient.sql("SELECT * FROM t_user WHERE name = :name AND email = :email")
                .bind("name", name)
                .bind("email", email)
                .map(this::mapRowToUser)
                .all();
    }

    @Override
    public Flux<User> findUsersByNameAndEmail2(String name, String email) {
        return databaseClient.sql("SELECT * FROM t_user WHERE name = $1 AND email = $2")
                .bind(0, name)
                .bind(1, email)
                .map(this::mapRowToUser)
                .all();
    }

    @Override
    public Flux<Long> batchInsertUsers(List<User> users) {
        return Flux.fromIterable(users)
                .flatMap(user -> databaseClient.sql("INSERT INTO t_user(id, name, age, email, balance, active) VALUES(:id, :name, :age, :email, :balance, :active)")
                        .bind("id", user.getId())
                        .bind("name", user.getName())
                        .bind("age", user.getAge())
                        .bind("email", user.getEmail())
                        .bind("balance", user.getBalance())
                        .bind("active", user.getActive())
                        .fetch()
                        .rowsUpdated()
                );
    }

    @Override
    public Mono<Long> batchUpdateUsers(List<User> users) {
        return Flux.fromIterable(users)
                .flatMap(user -> databaseClient.sql("UPDATE t_user SET name = :name, age = :age, email = :email, balance = :balance, active = :active WHERE id = :id")
                        .bind("name", user.getName())
                        .bind("age", user.getAge())
                        .bind("email", user.getEmail())
                        .bind("balance", user.getBalance())
                        .bind("active", user.getActive())
                        .bind("id", user.getId())
                        .fetch()
                        .rowsUpdated()
                )
                .reduce(Long::sum);
    }

    @Override
    public Flux<Void> executeInTransaction(Long fromId, Long toId, BigDecimal amount) {
        return transactionalOperator.execute(status -> databaseClient.sql("UPDATE t_user SET balance = balance - :amount WHERE id = :fromId")
                .bind("amount", amount)
                .bind("fromId", fromId)
                .fetch()
                .rowsUpdated()
                .then(databaseClient.sql("UPDATE t_user SET balance = balance + :amount WHERE id = :toId")
                        .bind("amount", amount)
                        .bind("toId", toId)
                        .fetch()
                        .rowsUpdated()
                )
                .then()
        );
    }

    @Override
    public Flux<Integer> batchInsertWithTransaction(List<User> users) {
        return transactionalOperator.execute(status -> Flux.fromIterable(users)
                .flatMap(user -> databaseClient.sql("INSERT INTO t_user(id, name, age, email, balance, active) VALUES(:id, :name, :age, :email, :balance, :active)")
                        .bind("id", user.getId())
                        .bind("name", user.getName())
                        .bind("age", user.getAge())
                        .bind("email", user.getEmail())
                        .bind("balance", user.getBalance())
                        .bind("active", user.getActive())
                        .fetch()
                        .rowsUpdated()
                )
                .reduce(Long::sum)
                .thenReturn(users.size())
        );
    }

    private User mapRowToUser(Row row, RowMetadata rowMetadata) {
        User user = new User();
        user.setId(row.get("id", Long.class));
        user.setName(row.get("name", String.class));
        user.setAge(row.get("age", Integer.class));
        user.setEmail(row.get("email", String.class));
        user.setBalance(new BigDecimal(row.get("balance", String.class)));
        user.setActive(row.get("active", Boolean.class));
        return user;
    }

    public static class UserRowMapper implements BiFunction<Row, RowMetadata, User> {
        @Override
        public User apply(Row row, RowMetadata rowMetadata) {
            User user = new User();
            user.setId(row.get("id", Long.class));
            user.setName(row.get("name", String.class));
            user.setAge(row.get("age", Integer.class));
            user.setEmail(row.get("email", String.class));
            user.setBalance(new BigDecimal(row.get("balance", String.class)));
            user.setActive(row.get("active", Boolean.class));
            return user;
        }
    }

    // Row到User的映射函数
    private final BiFunction<Row, RowMetadata, User> userRowMapper = (row, rowMetadata) -> {
        User user = new User();
        user.setId(row.get("id", Long.class));
        user.setName(row.get("name", String.class));
        user.setAge(row.get("age", Integer.class));
        user.setEmail(row.get("email", String.class));
        user.setBalance(new BigDecimal(row.get("balance", String.class)));
        user.setActive(row.get("active", Boolean.class));
        return user;
    };
}
